/*
 * BrokerClienteReceptor.cpp
 *
 *  Created on: Jan 28, 2010
 *      Author: jgrande
 */

#include "BrokerCliente.h"

void _brokerClienteReceptor(TCPSocket sock);
void contestarHandshake(const PDUPuntoAPunto& pdu, TCPSocket sock);
void encolarDatos(const PDUPuntoAPunto& pdu, TCPSocket sock);
void lanzarAnillo(const uint8_t* buff);
void agregarReceptor(const uint8_t* buff);
void quitarReceptor(const uint8_t* buff);
void terminarReceptor(Direccion dirRemota);

void receptor_sigint_handler(int signal) {
	while(wait(NULL)!=-1 || errno!=ECHILD);
	exit(0);
}

pid_t brokerClienteReceptor(TCPSocket sock) {
	pid_t pid;

	if((pid=fork())==0) {
		logger.set_pname("Middleware/Receptor");
		Utiles::registrar_signal(SIGINT, receptor_sigint_handler);
		_brokerClienteReceptor(sock);
		exit(0);
	}

	return pid;
}

void _brokerClienteReceptor(TCPSocket sock) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();
	PDUPuntoAPunto pdu;
	Direccion dirRemota = DIR_INVALIDA;

	logger << LogType::LOG_DEBUG_0 << "arrancando..." << logger.endl;

	forever {
		logger << LogType::LOG_DEBUG_0 <<"Esperando en socket..." << logger.endl;
		if(!sock.recv(&pdu)) {
			logger << LogType::LOG_ERROR <<"Error en TCPSocket.recv" << logger.err << logger.endl;
			if(dirRemota!=DIR_INVALIDA) {
				terminarReceptor(dirRemota);
			}
			sock.close();
			exit(1);
		}

		// imprime todos los campos del mensaje
		logger << LogType::LOG_DEBUG_0 << "Mensaje recibido: \n" <<
				"\tProtocolo:" << pdu.protocolo[0] << pdu.protocolo[1] << "\n" <<
				"\tVersion:" << (unsigned int)pdu.version << "\n" <<
				"\tOperacion:" << (unsigned int)pdu.operacion << "\n" <<
				"\tDestino:" << pdu.destino << "\n" <<
				"\tOrigen:" << pdu.origen << "\n" <<
				"\tLongitud datos:" << pdu.ldatos << logger.endl;

		// según el tipo de mensaje se realizan distintas acciones
		switch(pdu.operacion) {
		case middleware::common::OP_HANDSHAKE_REQ:
			dirRemota = pdu.origen;
			contestarHandshake(pdu, sock);
			if(brokerClienteEmisor(sock, pdu.origen)==(pid_t)-1) {
				logger << LogType::LOG_ERROR << "Error lanzando emisor" << logger.endl;
				sock.close();
				mw.getSem().signal();
				exit(1);
			}
			break;
		case middleware::common::OP_HANDSHAKE_RES:
			logger << LogType::LOG_DEBUG_0 << "hanshake completo, lanzando emisor" << logger.endl;
			if(brokerClienteEmisor(sock, pdu.origen)==(pid_t)-1) {
				logger << LogType::LOG_ERROR << "Error lanzando emisor" << logger.endl;
				sock.close();
				mw.getSem().signal();
				exit(1);
			}
			break;
		case middleware::common::OP_DATA:
			encolarDatos(pdu, sock);
			break;
		case middleware::common::OP_NUEVO_ANILLO:
			lanzarAnillo(pdu.datos);
			break;
		case middleware::common::OP_AGREGAR_RECEPTOR:
			agregarReceptor(pdu.datos);
			break;
		case middleware::common::OP_QUITAR_RECEPTOR:
			quitarReceptor(pdu.datos);
			break;
		default:
			logger << LogType::LOG_ERROR <<"Operación inválida (" << (int)pdu.operacion << ")" << logger.endl;
		}
	}
}

void contestarHandshake(const PDUPuntoAPunto& pdu, TCPSocket sock) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();
	Direccion dirLocal = ConfigSingleton::instance().getDirLocal();

	PDUPuntoAPunto newpdu = {
			{'P', 'P'}, 							// protocolo
			1, 										// version
			pdu.origen,								// destino
			dirLocal,								// origen
			middleware::common::OP_HANDSHAKE_RES,	// operación
			0,										// número de secuencia
			0,										// longitud datos
			{0}										// datos
	};

	logger << LogType::LOG_DEBUG_0 << "Contestando handshake" << logger.endl;

	if(!sock.send(&newpdu)) {
		logger << LogType::LOG_ERROR << "Error en TCPSocket.send" << logger.err << logger.endl;
		mw.getSem().signal();
		sock.close();
		exit(1);
	}
}

void encolarDatos(const PDUPuntoAPunto& pdu, TCPSocket sock) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();

	logger << LogType::LOG_DEBUG_0 <<"Encolando mensaje de datos" << logger.endl;

	long mtype = SECUENCIA_UNICAST;
	if(pdu.destino==DIR_MULTICAST) {
		mtype = (long)pdu.secuencia;
		logger << LogType::LOG_DEBUG_0 <<"recibido mensaje multicast numero " << mtype << logger.endl;
	}

	mw.getRcv().send((PDUPuntoAPunto*)&pdu, mtype);
	if(!mw.getRcv()) {
		logger << LogType::LOG_ERROR <<"Error en MessageQueue.send" << logger.err << logger.endl;
		sock.close();
		exit(0);
	}
}

void lanzarAnillo(const uint8_t* buff) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();
	MensajeNuevoAnillo msj;
	msj.loadFromArray(buff);

	logger << LogType::LOG_DEBUG_0 <<"lanzando anillo" << logger.endl;
	logger << LogType::LOG_DEBUG_0 << msj.toString() << logger.endl;

	mw.getListaReceptores().lock();
	list<Direccion>::const_iterator it;
	for(it=msj.getReceptores().begin();it!=msj.getReceptores().end();it++) {
		mw.getListaReceptores().add(*it);
	}
	mw.getListaReceptores().unlock();

	GestorAnillo anillo(msj.getSucesor(), (bool)msj.getPrimero());
	anillo.run();
}

void agregarReceptor(const uint8_t* buff) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();
	MensajeAgregarReceptor msj;
	msj.loadFromArray(buff);

	logger << "el host " << msj.getDirReceptor() << " fue agregado a "
			"la lista de receptores" << logger.endl;

	mw.getListaReceptores().lock();
	mw.getListaReceptores().add(msj.getDirReceptor());
	mw.getListaReceptores().unlock();
}

void quitarReceptor(const uint8_t* buff) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();
	MensajeQuitarReceptor msj;
	msj.loadFromArray(buff);

	logger << "el host " << msj.getDirReceptor() << " fue eliminado de "
			"la lista de receptores" << logger.endl;

	mw.getListaReceptores().lock();
	mw.getListaReceptores().remove(msj.getDirReceptor());
	mw.getListaReceptores().unlock();
}

void terminarReceptor(Direccion dirRemota) {
	MiddlewareCliente& mw = MiddlewareClienteSingleton::instance();

	logger << LogType::LOG_DEBUG_0 << "terminando receptor con host " << dirRemota << logger.endl;

	mw.getLineas().lock();

	ParEmisorReceptor pids = mw.getLineas().get(dirRemota);

	logger << LogType::LOG_DEBUG_0 << "interrumpiendo proceso emisor" << logger.endl;
	kill(pids.first, SIGINT);
	while(waitpid(pids.first, NULL, 0)!=pids.first);

	logger << LogType::LOG_DEBUG_0 << "emisor terminado, eliminando línea TCP " << logger.endl;
	mw.getLineas().remove(dirRemota);

	mw.getLineas().unlock();
}
